package cn.edu.flink.scala.tutorial.join

import cn.edu.flink.scala.tutorial.source.Event
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object IntervalJoinTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val orderStream = env.fromElements(
      ("Mary", "order-1", 5000L),
      ("Bob", "order-2", 5000L)
    ).assignAscendingTimestamps(_._3)

    val pvStream = env.fromElements(
      Event("Bob", "./cart", 2000L),
      Event("Bob", "./home", 3000L),
      Event("Bob", "./home", 4000L),
      Event("Bob", "./prod?id=1", 5000L),
      Event("Bob", "./prod?id=2", 6000L),
      Event("Bob", "./prod?id=3", 7000L)
    ).assignAscendingTimestamps(_.timestamp)


    orderStream
      .keyBy(_._1)
      .intervalJoin(pvStream.keyBy(_.user))
      .between(Time.seconds(-2), Time.seconds(0))
      .process(new ProcessJoinFunction[(String, String, Long), Event, String] {
        override def processElement(left: (String, String, Long), right: Event, ctx: ProcessJoinFunction[(String, String, Long), Event, String]#Context, out: Collector[String]): Unit = {
          out.collect(left + ": " + right)
        }
      }).print()

    env.execute("IntervalJoinTest")
  }
}
